/*
 *
 *  * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 *
 *
 */

package com.iwindplus.base.kafka.support;

import cn.hutool.core.collection.CollUtil;
import com.iwindplus.base.kafka.domain.property.KafkaProperty;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;

/**
 * kafka动态注册topic.
 *
 * @author zengdegui
 * @since 2025/11/22 10:53
 */
@Slf4j
public class KafkaTopicAutoCreator {

    /**
     * 幂等创建 Topic.
     *
     * @param adminClient 已配置好的 AdminClient
     * @param configs     期望的 Topic 配置
     * @param timeoutSec  等待超时（单位：秒，默认10）
     * @return Set<String>
     */
    public static Set<String> createTopicsIfAbsent(AdminClient adminClient, List<KafkaProperty.Config> configs, Long timeoutSec) {
        if (CollUtil.isEmpty(configs)) {
            return Set.of();
        }

        try {
            Set<String> exist = adminClient.listTopics().names().get(timeoutSec == null ? 10L : timeoutSec, TimeUnit.SECONDS);
            Set<NewTopic> newTopics = configs.stream()
                .filter(c -> !exist.contains(c.getTopicName()))
                .map(c -> {
                    int partitions = Optional.ofNullable(c.getNumPartitions()).filter(i -> i > 0).orElse(-1);
                    short replicationFactor = Optional.ofNullable(c.getReplicationFactor()).filter(i -> i > 0).orElse((short) -1);
                    NewTopic nt = new NewTopic(c.getTopicName(), partitions, replicationFactor);
                    Optional.ofNullable(c.getArguments()).filter(CollUtil::isNotEmpty).ifPresent(nt::configs);
                    return nt;
                })
                .collect(Collectors.toSet());

            if (CollUtil.isEmpty(newTopics)) {
                log.info("All topics already exist, skip create.");
                return Set.of();
            }

            CreateTopicsResult result = adminClient.createTopics(newTopics);
            final Set<String> val = result.values().keySet();
            log.info("Kafka topics created: {}", val);
            return val;
        } catch (Exception e) {
            log.warn("Kafka topics create failed: {}", e.getMessage(), e);
        }

        return Set.of();
    }
}
